package lww231;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;

//      k3      v3      k4      v4
public class MovieReducer extends Reducer<Text, MovieInfo, MovieInfo, NullWritable> {
    @Override
    protected void reduce(Text k3, Iterable<MovieInfo> v3, Context context)
            throws IOException, InterruptedException {
        List<MovieInfo> list = new ArrayList<>();
        for (MovieInfo movieInfo : v3) {
            MovieInfo info = new MovieInfo(movieInfo.getTitle(), movieInfo.getYear(),
                    movieInfo.getType(), movieInfo.getStar(),movieInfo.getDirector(),
                    movieInfo.getActor(),movieInfo.getTime(),movieInfo.getFilm_page());
            list.add(info);
        }
        Collections.sort(list, new Comparator<MovieInfo>() {

            @Override
            public int compare(MovieInfo o1, MovieInfo o2) {
                // TODO Auto-generated method stub
                if (o2.getStar() > o1.getStar()){
                    return 1;
                }else if (o2.getStar() < o1.getStar()){
                    return -1;
                }
                //如果star评分相同，按照year年份排序
                if (o2.getYear() > o1.getYear()){
                    return 1;
                }else {
                    return -1;
                }
            }
        });
        for (int i = 0; i < Math.min(5, list.size()); i++) {
            context.write(list.get(i), NullWritable.get());
        }
    }
}
